package com.chenjl.producer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.chenjl.Constants;
/**
 * 生产者
 * new Client API Producer,Comsumer 都直连Broker，由Broker和zk维护comsumer的offset
 * 2016-9-13 16:39:33
 * @author chenjinlong
 */
public class Producer02 {
	private static final Logger logger = LoggerFactory.getLogger(Producer02.class);
	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		Properties props = new Properties();
		props.put("bootstrap.servers",Constants.BROKER_URL);
		props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
		//
		props.put("acks","all");
	    props.put("retries",0);
	    
		logger.info("send begin");
		
		Producer<String,String> producer = new KafkaProducer<String,String>(props);
		
		ProducerRecord<String, String> hello= new ProducerRecord<String, String>(Constants.TOPIC_NAME,"hello key","hello value");
		//执行send时，并不是直接就向socket发起网络请求，而是先把数据存储到发送的缓冲区中，这个缓冲区的实现是一个RecordAccumulator实例
		producer.send(hello,new Callback() {
			@Override
			public void onCompletion(RecordMetadata metadata,Exception exception) {
				logger.info("hello message , offset : {} , topic : {}",metadata.offset(),metadata.topic());
			}
		});
		
		for(int i=1;i<=10;i++) {
			String messageKey = "message k :"+i;
			String messageValue = "value v : "+i;
			ProducerRecord<String, String> message = new ProducerRecord<String, String>(Constants.TOPIC_NAME,messageKey,messageValue);
			
			//异步发送, 同步等待
			Future<RecordMetadata> future = producer.send(message);
			RecordMetadata recordMetadata = future.get();
			logger.info("offset : {} , topic : {} , partition : {}",recordMetadata.offset(),recordMetadata.topic(),recordMetadata.partition());
		}
		logger.info("send end");
		
		producer.close();
	}
}